/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb.test;

import org.apache.log4j.PropertyConfigurator;

import com.qloudgen.csb.client.KfkProducer;

public class Server1 extends ServerBase {  
	
	private KfkProducer kp;
	private String topic;
	private String message;

	private Server1(String zkConn, String topic, int ackMode,
			int msgSize, int perfInterval) {
		super(perfInterval);
		
		this.topic = topic;
		message = Utils.randomString(msgSize);
		kp = new KfkProducer(zkConn, ackMode, null);
	}
	
	public void send() {
		kp.send(topic, message);
		count++;
	}

    public static void main(String[] args) {
    	PropertyConfigurator.configure("conf/log4j.properties");
    	
    	final String zkConn = args[0];
    	final String topic = args[1];
    	final int ackMode = Integer.parseInt(args[2]);
    	final int msgSize = Integer.parseInt(args[3]);    	
    	final int nThreads = Integer.parseInt(args[4]);
    	final int perfInterval = Integer.parseInt(args[5]);
    	
    	final Sum isum = new Sum(nThreads);
    	for (int i = 0; i < nThreads; i++) {
    		new Thread(new Runnable() {    		
    			@Override
    			public void run() {
    				Server1 srv = new  Server1(zkConn, topic,
    					ackMode, msgSize, perfInterval);
    				srv.perf(isum);
    	
    				while (!Thread.currentThread().isInterrupted()) {
    					srv.send();
    				}  
    			}
    		}).start();
    	}
    	
    	while (!Thread.currentThread().isInterrupted()) {
    		try {
    			System.out.println("server1 running with " 
    					+ nThreads + " thread...");
				Thread.sleep(Integer.MAX_VALUE);
			} catch (InterruptedException e) {}
    	}    	
    	System.out.println("server1 exiting...");
    	System.exit(0);
    }
}

